package com.hkbigdata.streamCoreCoding;

import com.hkbigdata.entry.MarketingUserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @author liuanbo
 * @creat 2023-04-10-14:36
 * @see 2194550857@qq.com
 */
public class Flink05_NoMarketByChannel {
    public static void main(String[] args) throws Exception {
        //1.运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.获取数据
        DataStreamSource<MarketingUserBehavior> marketingUserBehaviorDataStreamSource = env.addSource(new AppMarketingDataSource());

        //3.转换位Tuple2
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = marketingUserBehaviorDataStreamSource.map(new MapFunction<MarketingUserBehavior, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(MarketingUserBehavior value) throws Exception {
                return new Tuple2<>(value.getBehavior(), 1);
            }
        });

        map.keyBy(data -> data.f0).sum(1).print();


        env.execute();

    }

    public static class AppMarketingDataSource extends RichSourceFunction<MarketingUserBehavior> {
        //随机变量
        private Random random = new Random();
        //准备渠道和操作行为数据
        private List<String> channels = Arrays.asList("huawei", "xiaomi", "taobao", "vivo", "appo", "jidong");
        private List<String> behaviors = Arrays.asList("download", "install", "updated", "uninstall");
        //循环条件
        private Boolean scan = true;

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (scan) {
                //将渠道，用户行为等数据转换为javabean
                MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                        (long) random.nextInt(1000000),
                        behaviors.get(random.nextInt(behaviors.size())),
                        channels.get(random.nextInt(channels.size())),
                        System.currentTimeMillis()
                );
                //输出对象
                ctx.collect(marketingUserBehavior);
                Thread.sleep(1000);
            }

        }

        @Override
        public void cancel() {

        }
    }
}
